/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to you under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.calcite.test.enumerable;

import org.apache.calcite.adapter.enumerable.EnumerableMergeJoin;
import org.apache.calcite.adapter.enumerable.EnumerableRules;
import org.apache.calcite.adapter.java.ReflectiveSchema;
import org.apache.calcite.config.CalciteConnectionProperty;
import org.apache.calcite.config.Lex;
import org.apache.calcite.interpreter.Bindables;
import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.runtime.Hook;
import org.apache.calcite.sql.fun.SqlStdOperatorTable;
import org.apache.calcite.test.CalciteAssert;
import org.apache.calcite.test.schemata.hr.HierarchySchema;
import org.apache.calcite.test.schemata.hr.HrSchema;
import org.apache.calcite.test.schemata.hr.HrSchemaBig;

import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.function.Consumer;

/**
 * Unit tests for the different Enumerable Join implementations.
 */
class EnumerableJoinTest {

  /** Test case for
   * <a href="https://issues.apache.org/jira/browse/CALCITE-2968">[CALCITE-2968]
   * New AntiJoin relational expression</a>. */
  @Test void equiAntiJoin() {
    tester(false, new HrSchema())
        .withRel(
            // Retrieve departments without employees. Equivalent SQL:
            //   SELECT d.deptno, d.name FROM depts d
            //   WHERE NOT EXISTS (SELECT 1 FROM emps e WHERE e.deptno = d.deptno)
            builder -> builder
                .scan("s", "depts").as("d")
                .scan("s", "emps").as("e")
                .antiJoin(
                    builder.equals(
                        builder.field(2, "d", "deptno"),
                        builder.field(2, "e", "deptno")))
                .project(
                    builder.field("deptno"),
                    builder.field("name"))
                .build())
        .returnsUnordered(
            "deptno=30; name=Marketing",
            "deptno=40; name=HR");
  }

  @Test void asofJoinTest() {
    tester(false, new HrSchema())
        .withRel(
            // select d.deptno, e.empid from emps e left asof join depts d
            // match_condition d.name <= e.name
            // on d.deptno = e.deptno
            builder -> builder
                .scan("s", "depts").as("d")
                .scan("s", "emps").as("e")
                .asofJoin(JoinRelType.LEFT_ASOF,
                    builder.equals(
                        builder.field(2, "d", "deptno"),
                        builder.field(2, "e", "deptno")),
                    builder.lessThan(
                        builder.field(2, "d", "name"),
                        builder.field(2, "e", "name")))
                .project(
                    builder.field("deptno"),
                    builder.field("e", "name"),
                    builder.field("empid"))
                .build())
        .returnsUnordered(
            "deptno=10; name=Theodore; empid=110",
            "deptno=30; name=null; empid=null",
            "deptno=40; name=null; empid=null");

    tester(false, new HrSchema())
        .withRel(
            // select d.deptno, e.empid from emps e asof join depts d
            // match_condition e.name <= d.name
            // on d.deptno = e.deptno
            builder -> builder
                .scan("s", "depts").as("d")
                .scan("s", "emps").as("e")
                .asofJoin(JoinRelType.ASOF,
                    builder.equals(
                        builder.field(2, "d", "deptno"),
                        builder.field(2, "e", "deptno")),
                    builder.lessThan(
                        builder.field(2, "e", "name"),
                        builder.field(2, "d", "name")))
                .project(
                    builder.field("deptno"),
                    builder.field("e", "name"),
                    builder.field("empid"))
                .build())
        .returnsUnordered(
            "deptno=10; name=Bill; empid=100");
  }

  /** Test case for
   * <a href="https://issues.apache.org/jira/browse/CALCITE-2968">[CALCITE-2968]
   * New AntiJoin relational expression</a>. */
  @Test void nonEquiAntiJoin() {
    tester(false, new HrSchema())
        .withRel(
            // Retrieve employees with the top salary in their department. Equivalent SQL:
            //   SELECT e.name, e.salary FROM emps e
            //   WHERE NOT EXISTS (
            //     SELECT 1 FROM emps e2
            //     WHERE e.deptno = e2.deptno AND e2.salary > e.salary)
            builder -> builder
                .scan("s", "emps").as("e")
                .scan("s", "emps").as("e2")
                .antiJoin(
                    builder.and(
                        builder.equals(
                            builder.field(2, "e", "deptno"),
                            builder.field(2, "e2", "deptno")),
                        builder.call(
                            SqlStdOperatorTable.GREATER_THAN,
                            builder.field(2, "e2", "salary"),
                            builder.field(2, "e", "salary"))))
                .project(
                    builder.field("name"),
                    builder.field("salary"))
                .build())
        .returnsUnordered(
            "name=Theodore; salary=11500.0",
            "name=Eric; salary=8000.0");
  }

  /** Test case for
   * <a href="https://issues.apache.org/jira/browse/CALCITE-2968">[CALCITE-2968]
   * New AntiJoin relational expression</a>. */
  @Test void equiAntiJoinWithNullValues() {
    final Integer salesDeptNo = 10;
    tester(false, new HrSchema())
        .withRel(
            // Retrieve employees from any department other than Sales (deptno 10) whose
            // commission is different from any Sales employee commission. Since there
            // is a Sales employee with null commission, the goal is to validate that antiJoin
            // behaves as a NOT EXISTS (and returns results), and not as a NOT IN (which would
            // not return any result due to its null handling). Equivalent SQL:
            //   SELECT empOther.empid, empOther.name FROM emps empOther
            //   WHERE empOther.deptno <> 10 AND NOT EXISTS
            //     (SELECT 1 FROM emps empSales
            //      WHERE empSales.deptno = 10 AND empSales.commission = empOther.commission)
            builder -> builder
                .scan("s", "emps").as("empOther")
                .filter(
                    builder.notEquals(
                        builder.field("empOther", "deptno"),
                        builder.literal(salesDeptNo)))
                .scan("s", "emps").as("empSales")
                .filter(
                    builder.equals(
                        builder.field("empSales", "deptno"),
                        builder.literal(salesDeptNo)))
                .antiJoin(
                    builder.equals(
                        builder.field(2, "empOther", "commission"),
                        builder.field(2, "empSales", "commission")))
                .project(
                    builder.field("empid"),
                    builder.field("name"))
                .build())
        .returnsUnordered("empid=200; name=Eric");
  }

  /** Test case for
   * <a href="https://issues.apache.org/jira/browse/CALCITE-3170">[CALCITE-3170]
   * ANTI join on conditions push down generates wrong plan</a>. */
  @Test void testCanNotPushAntiJoinConditionsToLeft() {
    tester(false, new HrSchema())
        .withRel(
            // build a rel equivalent to sql:
            // select * from emps
            // where emps.deptno
            // not in (select depts.deptno from depts where emps.name = 'ddd')

            // Use `equals` instead of `is not distinct from` only for testing.
            builder -> builder
                .scan("s", "emps")
                .scan("s", "depts")
                .antiJoin(
                    builder.equals(
                        builder.field(2, 0, "deptno"),
                        builder.field(2, 1, "deptno")),
                    builder.equals(builder.field(2, 0, "name"),
                        builder.literal("ddd")))
                .project(builder.field(0))
                .build())
        .returnsUnordered(
            "empid=100",
            "empid=110",
            "empid=150",
            "empid=200");
  }

  /**
   * The test verifies if {@link EnumerableMergeJoin} can implement a join with non-equi conditions.
   */
  @Test void testSortMergeJoinWithNonEquiCondition() {
    tester(false, new HrSchema())
        .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner -> {
          planner.addRule(EnumerableRules.ENUMERABLE_MERGE_JOIN_RULE);
          planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE);
        })
        .withRel(builder -> builder
            // build a rel equivalent to sql:
            // select e.empid, e.name, d.name as dept, e.deptno, d.deptno
            // from emps e join depts d
            // on e.deptno=d.deptno and e.empid > d.deptno * 10
            // Note: explicit sort is used so EnumerableMergeJoin could actually work
            .scan("s", "emps")
            .sort(builder.field("deptno"))
            .scan("s", "depts")
            .sort(builder.field("deptno"))
            .join(JoinRelType.INNER,
                builder.and(
                    builder.equals(
                        builder.field(2, 0, "deptno"),
                        builder.field(2, 1, "deptno")),
                    builder.getRexBuilder().makeCall(
                        SqlStdOperatorTable.GREATER_THAN,
                        builder.field(2, 0, "empid"),
                        builder.getRexBuilder().makeCall(
                            SqlStdOperatorTable.MULTIPLY,
                            builder.literal(10),
                            builder.field(2, 1, "deptno")))))
            .project(
                builder.field(1, "emps", "empid"),
                builder.field(1, "emps", "name"),
                builder.alias(builder.field(1, "depts", "name"), "dept_name"),
                builder.alias(builder.field(1, "emps", "deptno"), "e_deptno"),
                builder.alias(builder.field(1, "depts", "deptno"), "d_deptno"))
            .build())
        .explainHookMatches("" // It is important that we have MergeJoin in the plan
            + "EnumerableCalc(expr#0..4=[{inputs}], empid=[$t0], name=[$t2], dept_name=[$t4], e_deptno=[$t1], d_deptno=[$t3])\n"
            + "  EnumerableMergeJoin(condition=[AND(=($1, $3), >($0, *(10, $3)))], joinType=[inner])\n"
            + "    EnumerableSort(sort0=[$1], dir0=[ASC])\n"
            + "      EnumerableCalc(expr#0..4=[{inputs}], proj#0..2=[{exprs}])\n"
            + "        EnumerableTableScan(table=[[s, emps]])\n"
            + "    EnumerableSort(sort0=[$0], dir0=[ASC])\n"
            + "      EnumerableCalc(expr#0..3=[{inputs}], proj#0..1=[{exprs}])\n"
            + "        EnumerableTableScan(table=[[s, depts]])\n")
        .returnsUnordered("empid=110; name=Theodore; dept_name=Sales; e_deptno=10; d_deptno=10",
            "empid=150; name=Sebastian; dept_name=Sales; e_deptno=10; d_deptno=10");
  }

  /** Test case for
   * <a href="https://issues.apache.org/jira/browse/CALCITE-3846">[CALCITE-3846]
   * EnumerableMergeJoin: wrong comparison of composite key with null values</a>. */
  @Test void testMergeJoinInnerWithCompositeKeyAndNullValues() {
    checkMergeJoinWithCompositeKeyAndNullValues(
        false,
        JoinRelType.INNER,
        "empid=110; empid0=110",
        "empid=100; empid0=100",
        "empid=200; empid0=200");
    checkMergeJoinWithCompositeKeyAndNullValues(
        true,
        JoinRelType.INNER,
        "empid=48; empid0=48",
        "empid=4; empid0=4",
        "empid=4; empid0=8");
  }

  @Test void testMergeJoinLeftWithCompositeKeyAndNullValues() {
    checkMergeJoinWithCompositeKeyAndNullValues(
        false,
        JoinRelType.LEFT,
        "empid=110; empid0=110",
        "empid=100; empid0=100",
        "empid=150; empid0=null",
        "empid=200; empid0=200");
    checkMergeJoinWithCompositeKeyAndNullValues(
        true,
        JoinRelType.LEFT,
        "empid=48; empid0=48",
        "empid=47; empid0=null",
        "empid=4; empid0=4");
  }

  @Test void testMergeJoinSemiWithCompositeKeyAndNullValues() {
    checkMergeJoinWithCompositeKeyAndNullValues(
        false,
        JoinRelType.SEMI,
        "empid=110",
        "empid=100",
        "empid=200");
    checkMergeJoinWithCompositeKeyAndNullValues(
        true,
        JoinRelType.SEMI,
        "empid=48",
        "empid=4",
        "empid=8");
  }

  @Test void testMergeJoinAntiWithCompositeKeyAndNullValues() {
    checkMergeJoinWithCompositeKeyAndNullValues(
        false,
        JoinRelType.ANTI,
        "empid=150");
    checkMergeJoinWithCompositeKeyAndNullValues(
        true,
        JoinRelType.ANTI,
        "empid=47",
        "empid=3",
        "empid=7");
  }

  private void checkMergeJoinWithCompositeKeyAndNullValues(boolean bigSchema, JoinRelType joinType,
      String... expected) {
    CalciteAssert.AssertQuery checker =
        tester(false, bigSchema ? new HrSchemaBig() : new HrSchema())
        .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner -> {
          planner.addRule(EnumerableRules.ENUMERABLE_MERGE_JOIN_RULE);
          planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE);
        })
        .withRel(builder -> builder
            .scan("s", "emps").as("e1")
            .sort(builder.field("deptno"), builder.field("commission"), builder.field("empid"))
            .scan("s", "emps").as("e2")
            .sort(builder.field("deptno"), builder.field("commission"), builder.field("empid"))
            .join(joinType,
                builder.and(
                    builder.equals(
                        builder.field(2, 0, "deptno"),
                        builder.field(2, 1, "deptno")),
                    builder.equals(
                        builder.field(2, 0, "commission"),
                        builder.field(2, 1, "commission"))))
            .project(joinType.projectsRight()
                ? Arrays.asList(builder.field("e1", "empid"), builder.field("e2", "empid"))
                : Arrays.asList(builder.field("e1", "empid")))
            .build())
        .explainHookContains("EnumerableMergeJoin"); // We must have MergeJoin in the plan
    if (bigSchema) {
      checker.returnsStartingWith(expected);
    } else {
      checker.returnsOrdered(expected);
    }
  }

  /** Test case for
   * <a href="https://issues.apache.org/jira/browse/CALCITE-3820">[CALCITE-3820]
   * EnumerableDefaults#orderBy should be lazily computed + support enumerator
   * re-initialization</a>. */
  @Test void testRepeatUnionWithMergeJoin() {
    tester(false, new HierarchySchema())
        .withHook(Hook.PLANNER, (Consumer<RelOptPlanner>) planner -> {
          planner.addRule(Bindables.BINDABLE_TABLE_SCAN_RULE);
          planner.addRule(EnumerableRules.ENUMERABLE_MERGE_JOIN_RULE);
          planner.removeRule(EnumerableRules.ENUMERABLE_JOIN_RULE);
        })
        // Note: explicit sort is used so EnumerableMergeJoin can actually work
        .withRel(builder -> builder
            //   WITH RECURSIVE delta(empid, name) as (
            //     SELECT empid, name FROM emps WHERE empid = 2
            //     UNION ALL
            //     SELECT e.empid, e.name FROM delta d
            //                            JOIN hierarchies h ON d.empid = h.managerid
            //                            JOIN emps e        ON h.subordinateid = e.empid
            //   )
            //   SELECT empid, name FROM delta
            .scan("s", "emps")
            .filter(
                builder.equals(
                    builder.field("empid"),
                    builder.literal(2)))
            .project(
                builder.field("emps", "empid"),
                builder.field("emps", "name"))

            .transientScan("#DELTA#")
            .sort(builder.field("empid"))
            .scan("s", "hierarchies")
            .sort(builder.field("managerid"))
            .join(
                JoinRelType.INNER,
                builder.equals(
                    builder.field(2, "#DELTA#", "empid"),
                    builder.field(2, "hierarchies", "managerid")))
            .sort(builder.field("subordinateid"))

            .scan("s", "emps")
            .sort(builder.field("empid"))
            .join(
                JoinRelType.INNER,
                builder.equals(
                    builder.field(2, "hierarchies", "subordinateid"),
                    builder.field(2, "emps", "empid")))
            .project(
                builder.field("emps", "empid"),
                builder.field("emps", "name"))
            .repeatUnion("#DELTA#", true)
            .build()
        )
        .explainHookMatches("" // It is important to have MergeJoin + EnumerableSort in the plan
            + "EnumerableRepeatUnion(all=[true])\n"
            + "  EnumerableTableSpool(readType=[LAZY], writeType=[LAZY], table=[[#DELTA#]])\n"
            + "    EnumerableCalc(expr#0..4=[{inputs}], expr#5=[2], expr#6=[=($t0, $t5)], empid=[$t0], name=[$t2], $condition=[$t6])\n"
            + "      EnumerableTableScan(table=[[s, emps]])\n"
            + "  EnumerableTableSpool(readType=[LAZY], writeType=[LAZY], table=[[#DELTA#]])\n"
            + "    EnumerableCalc(expr#0..4=[{inputs}], empid=[$t3], name=[$t4])\n"
            + "      EnumerableMergeJoin(condition=[=($2, $3)], joinType=[inner])\n"
            + "        EnumerableSort(sort0=[$2], dir0=[ASC])\n"
            + "          EnumerableMergeJoin(condition=[=($0, $1)], joinType=[inner])\n"
            + "            EnumerableSort(sort0=[$0], dir0=[ASC])\n"
            + "              EnumerableCalc(expr#0..1=[{inputs}], empid=[$t0])\n"
            + "                EnumerableInterpreter\n"
            + "                  BindableTableScan(table=[[#DELTA#]])\n"
            + "            EnumerableSort(sort0=[$0], dir0=[ASC])\n"
            + "              EnumerableTableScan(table=[[s, hierarchies]])\n"
            + "        EnumerableSort(sort0=[$0], dir0=[ASC])\n"
            + "          EnumerableCalc(expr#0..4=[{inputs}], empid=[$t0], name=[$t2])\n"
            + "            EnumerableTableScan(table=[[s, emps]])\n")
        .returnsUnordered("empid=2; name=Emp2",
            "empid=3; name=Emp3",
            "empid=5; name=Emp5");
  }

  private CalciteAssert.AssertThat tester(boolean forceDecorrelate,
      Object schema) {
    return CalciteAssert.that()
        .with(CalciteConnectionProperty.LEX, Lex.JAVA)
        .with(CalciteConnectionProperty.FORCE_DECORRELATE, forceDecorrelate)
        .withSchema("s", new ReflectiveSchema(schema));
  }
}
